package com.data.sample.service.impl;

import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.admin.AdminClient;
import org.apache.kafka.clients.admin.CreateTopicsResult;
import org.apache.kafka.clients.admin.NewTopic;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaAdmin;
import org.springframework.stereotype.Service;

import java.util.Collections;
import java.util.concurrent.ExecutionException;

@Service
@Slf4j
public class KafkaTopicService {

    // 注入Spring自动配置的KafkaAdmin（已关联Kafka地址）
    @Autowired
    private KafkaAdmin kafkaAdmin;

    public void dynamicCreateTopic(String topic, int partitions, short replicas) {
        try {
            // 先判断Topic是否已存在，避免重复创建报错
            if (isTopicExists(topic)) {
                log.info("Topic [" + topic + "] 已存在，无需重复创建");
            }else
            {
                // 创建Topic
                createTopic(topic, partitions, replicas);
                log.info("Topic [" + topic + "] 创建成功");
            }
        } catch (Exception e) {
            e.printStackTrace();
            log.error("Topic [" + topic + "] 创建失败：" + e.getMessage());
        }
    }

    /**
     * 动态创建Kafka Topic
     * @param topicName Topic名称（动态传入，如用户输入）
     * @param partitions 分区数
     * @param replicas 副本数
     * @throws ExecutionException 执行异常（如Kafka连接失败）
     * @throws InterruptedException 线程中断异常
     */
    public void createTopic(String topicName, int partitions, short replicas)
            throws ExecutionException, InterruptedException {
        // 1. 获取AdminClient实例（操作Kafka的核心客户端）
        try (AdminClient adminClient = AdminClient.create(kafkaAdmin.getConfigurationProperties())) {
            // 2. 构建NewTopic对象
            NewTopic newTopic = new NewTopic(topicName, partitions, replicas);
            // 3. 调用createTopics方法创建Topic（需调用get()等待结果，否则是异步）
            CreateTopicsResult result = adminClient.createTopics(Collections.singleton(newTopic));
            result.all().get();  // 阻塞等待创建完成，确保Topic创建成功
            System.out.printf("Topic [%s] 动态创建成功（分区数：%d，副本数：%d）%n", 
                             topicName, partitions, replicas);
        }
    }

    /**
     * 可选：判断Topic是否已存在（避免重复创建报错）
     */
    public boolean isTopicExists(String topicName) throws ExecutionException, InterruptedException {
        try (AdminClient adminClient = AdminClient.create(kafkaAdmin.getConfigurationProperties())) {
            // 调用listTopics()获取所有Topic，判断目标Topic是否在列表中
            return adminClient.listTopics().names().get().contains(topicName);
        }
    }
}